package com.bigdata.assignment.problem3;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.*;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Counters;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import java.io.*;
import java.net.URI;
import java.util.*;

/**
 * MapReduce作业驱动类：优化版WordCount程序的入口
 * 功能：配置作业参数（分片大小、Reduce任务数、是否启用Combiner）、提交作业、处理输出结果、生成性能报告
 */
public class WordCountOptimizedDriver {

    public static void main(String[] args) throws Exception {
        // 验证输入参数数量（必须为5个，新增Combiner开关参数）
        if (args.length != 5) {
            System.err.println(
                    "Usage: WordCountOptimizedDriver <inputPath> <outputPath> <splitMaxSize> <numReduceTasks> <enableCombiner(true/false)>");
            System.err.println(
                    "Example: WordCountOptimizedDriver /public/data/wordcount /users/s123/homework1/problem3 67108864 4 true");
            System.exit(1);
        }

        // 从命令行参数解析配置项
        String INPUT = args[0];         // 输入数据在HDFS上的路径
        String OUTPUT = args[1];        // 输出结果在HDFS上的路径
        String splitSize = args[2];     // 分片最大大小（单位：字节，用于控制Map任务数量）
        int numReducers;                // Reduce任务数量（并行度控制）
        boolean enableCombiner;         // 是否启用Combiner（Map端本地聚合开关）

        // 解析Reduce任务数量：必须为正整数
        try {
            numReducers = Integer.parseInt(args[3]);
            if (numReducers < 1) {
                throw new IllegalArgumentException("Reduce任务数量必须大于0");
            }
        } catch (NumberFormatException e) {
            System.err.println("错误：Reduce任务数量必须是整数（参数4）");
            System.exit(1);
            return;
        } catch (IllegalArgumentException e) {
            System.err.println("错误：" + e.getMessage());
            System.exit(1);
            return;
        }

        // 解析Combiner开关参数：必须为true或false
        try {
            enableCombiner = Boolean.parseBoolean(args[4]);
        } catch (Exception e) {
            System.err.println("错误：Combiner开关参数必须为true或false（参数5）");
            System.exit(1);
            return;
        }

        // 初始化Hadoop配置对象
        Configuration conf = new Configuration();

        // 设置输入文件分片最大大小（控制Map任务数量：分片越小，Map任务越多）
        // 配置项mapreduce.input.fileinputformat.split.maxsize用于指定单个分片的最大字节数
        conf.set("mapreduce.input.fileinputformat.split.maxsize", splitSize);

        // 创建Job实例：指定作业名称为"Word Count Problem3 Optimized"
        Job job = Job.getInstance(conf, "Word Count Problem3 Optimized");
        // 设置作业的主类（当前类）
        job.setJarByClass(WordCountOptimizedDriver.class);

        // 配置Mapper和Reducer类（固定实现）
        job.setMapperClass(WordCountOptimizedMapper.class);    // 设置自定义Mapper
        job.setReducerClass(WordCountOptimizedReducer.class);  // 设置自定义Reducer

        // 根据参数决定是否启用Combiner（Combiner逻辑与Reducer一致时可复用）
        // Combiner作用：在Map端对输出结果进行本地聚合，减少Shuffle阶段的数据传输量
        if (enableCombiner) {
            job.setCombinerClass(WordCountOptimizedCombiner.class);  // 启用Combiner
            System.out.println("✅ 已开启Combiner");
        } else {
            System.out.println("ℹ️ 未开启Combiner");
        }

        // 设置作业输出的键值对类型（与Reducer输出一致）
        job.setOutputKeyClass(Text.class);          // 输出key类型：单词（Text）
        job.setOutputValueClass(IntWritable.class); // 输出value类型：计数（IntWritable）

        // 设置Reduce任务数量（通过参数动态配置）
        job.setNumReduceTasks(numReducers);

        // 初始化HDFS文件系统对象（用于检查输入输出路径）
        FileSystem fs = FileSystem.get(new URI(INPUT), conf);

        // 检查输入目录是否存在：若不存在则退出
        if (!fs.exists(new Path(INPUT))) {
            System.err.println("输入目录不存在：" + INPUT);
            return;
        }

        // 若输出目录已存在则删除：避免作业因目录存在而失败
        if (fs.exists(new Path(OUTPUT))) {
            fs.delete(new Path(OUTPUT), true);  // 递归删除目录及内容
        }

        // 设置作业的输入路径和输出路径
        FileInputFormat.addInputPath(job, new Path(INPUT));   // 添加输入路径
        FileOutputFormat.setOutputPath(job, new Path(OUTPUT)); // 设置输出路径

        // 记录作业提交开始时间（用于计算客户端总耗时）
        long start = System.currentTimeMillis();
        // 提交作业并等待完成：返回作业是否成功执行
        boolean success = job.waitForCompletion(true);
        // 记录作业完成时间
        long end = System.currentTimeMillis();

        // 计算客户端总耗时（包含作业提交、等待资源、执行的全流程时间）
        long clientTotalTime = end - start;

        // 计算集群实际执行时间（仅包含作业在集群上的运行时间，排除客户端等待资源的时间）
        // job.getStartTime()：作业开始执行的时间（集群分配资源后）
        // job.getFinishTime()：作业执行完成的时间
        long actualRunTime = job.getFinishTime() - job.getStartTime();

        // 若作业执行失败，输出错误信息并退出
        if (!success) {
            System.err.println("Job failed");
            return;
        }

        // 获取作业计数器（用于统计各种运行指标）
        Counters counters = job.getCounters();

        // 从计数器中提取关键指标
        long mapTasks = counters.findCounter("org.apache.hadoop.mapreduce.JobCounter", "TOTAL_LAUNCHED_MAPS")
                .getValue(); // 总Map任务数量
        long totalLines = counters.findCounter(WordCountOptimizedMapper.COUNTERS.INPUT_LINES).getValue(); // 输入总行数（输入记录数）
        long cleanedWords = counters.findCounter(WordCountOptimizedMapper.COUNTERS.CLEANED_WORDS).getValue(); // 清洗后的总单词数
        
        // 若启用Combiner，获取Combiner的输入/输出记录数；否则设为0
        long combIn = enableCombiner ? counters.findCounter(WordCountOptimizedCombiner.COUNTERS.COMBINER_INPUT).getValue() : 0;
        long combOut = enableCombiner ? counters.findCounter(WordCountOptimizedCombiner.COUNTERS.COMBINER_OUTPUT).getValue() : 0;
        
        long reducerOutput = counters.findCounter(WordCountOptimizedReducer.COUNTERS.REDUCER_OUTPUT_WORDS).getValue(); // Reducer输出的单词总数（去重后）
        long time = actualRunTime; // 集群实际运行时间

        // 读取Reducer输出结果（part-r-00000）并按词频降序排序
        List<String> lines = new ArrayList<>();
        FSDataInputStream inputStream = fs.open(new Path(OUTPUT + "/part-r-00000")); // 打开Reducer输出文件
        BufferedReader br = new BufferedReader(new InputStreamReader(inputStream));
        String line;
        while ((line = br.readLine()) != null) {
            lines.add(line); // 读取所有行到列表
        }
        br.close(); // 关闭输入流

        // 按单词出现次数降序排序（提取每行的计数部分进行比较）
        lines.sort((a, b) -> {
            int countA = Integer.parseInt(a.split("\t")[1]); // 分割字符串，获取计数
            int countB = Integer.parseInt(b.split("\t")[1]);
            return Integer.compare(countB, countA); // 降序排列（b - a）
        });

        // 将排序后的结果写入word-count-results.txt
        FSDataOutputStream out = fs.create(new Path(OUTPUT + "/word-count-results.txt"), true); // 创建输出流
        for (String l : lines) {
            out.writeBytes(l + "\n"); // 写入每行并换行
        }
        out.close(); // 关闭输出流

        // 生成性能报告（记录关键指标，用于后续性能分析）
        FSDataOutputStream perf = fs.create(new Path(OUTPUT + "/performance-report.txt"), true);
        perf.writeBytes("total_processing_time\t" + time + "\n"); // 总处理时间（集群实际运行时间）
        perf.writeBytes("client_total_time\t" + clientTotalTime + " ms\n"); // 客户端总耗时
        perf.writeBytes("actual_run_time\t" + actualRunTime + " ms\n"); // 集群实际运行时间
        perf.writeBytes("map_tasks_count\t" + mapTasks + "\n"); // Map任务数量
        perf.writeBytes("reduce_tasks_count\t" + numReducers + "\n"); // Reduce任务数量
        perf.writeBytes("input_records\t" + totalLines + "\n"); // 输入记录数（总行数）
        perf.writeBytes("output_records\t" + reducerOutput + "\n"); // 输出记录数（去重后单词数）
        perf.writeBytes("total_words\t" + cleanedWords + "\n"); // 总单词数（清洗后）
        perf.writeBytes("combiner_enabled\t" + enableCombiner + "\n"); // Combiner启用状态
        perf.writeBytes("combiner_input_records\t" + combIn + "\n"); // Combiner输入记录数
        perf.writeBytes("combiner_output_records\t" + combOut + "\n"); // Combiner输出记录数
        perf.close(); // 关闭性能报告输出流

        System.out.println("🎉 Problem 3 Completed Successfully!");
    }
}